package com.bw.gmall.realtime.ods;

import com.bw.gmall.realtime.bean.AdsDmpBasicFeature;
import com.bw.gmall.realtime.bean.DwsUserAgeFeature;
import com.bw.gmall.realtime.bean.OdsUserBasicInfo;
import com.bw.gmall.realtime.bean.OdsUserBehavior;
import lombok.Data;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import com.alibaba.fastjson.JSON;

import java.security.Timestamp;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

/**
 * 达摩盘用户画像基础特征分层计算Flink Job
 * 工单编号：大数据-用户画像-11-达摩盘基础特征
 */
public class DmpBasicFeatureFlinkJob {
    public static void main(String[] args) throws Exception {
        // 1. 初始化Flink执行环境（批流一体）
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        EnvironmentSettings tableEnvSettings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .build();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, tableEnvSettings);

        // 2. 配置Kafka数据源（模拟ODS层原始数据接入，文档中数据来源：淘宝行为日志、会员资料等）
        Properties kafkaProps = new Properties();
        kafkaProps.setProperty("bootstrap.servers", "kafka-node1:9092,kafka-node2:9092");
        kafkaProps.setProperty("group.id", "dmp_basic_feature_group");

        // 2.1 读取ODS层用户行为数据（Kafka Topic：ods_user_behavior）
        DataStream<String> odsUserBehaviorJsonStream = env.addSource(
                new FlinkKafkaConsumer<>("ods_user_behavior", new SimpleStringSchema(), kafkaProps)
        );
        DataStream<OdsUserBehavior> odsUserBehaviorStream = odsUserBehaviorJsonStream
                .map(new MapFunction<String, OdsUserBehavior>() {
                    @Override
                    public OdsUserBehavior map(String value) throws Exception {
                        return JSON.parseObject(value, OdsUserBehavior.class);
                    }
                });

        // 2.2 读取ODS层用户基础信息数据（Kafka Topic：ods_user_basic_info）
        DataStream<String> odsUserBasicInfoJsonStream = env.addSource(
                new FlinkKafkaConsumer<>("ods_user_basic_info", new SimpleStringSchema(), kafkaProps)
        );
        DataStream<OdsUserBasicInfo> odsUserBasicInfoStream = odsUserBasicInfoJsonStream
                .map(new MapFunction<String, OdsUserBasicInfo>() {
                    @Override
                    public OdsUserBasicInfo map(String value) throws Exception {
                        return JSON.parseObject(value, OdsUserBasicInfo.class);
                    }
                });

        // 3. DWD层：数据清洗与标准化（遵循文档中异常值过滤、单位统一规则）
        // 3.1 DWD层-用户行为清洗（匹配商品信息、搜索词分类）
        DataStream<DwdUserBehaviorClean> dwdUserBehaviorCleanStream = odsUserBehaviorStream
                .map(new DwdUserBehaviorCleanMap()); // 自定义MapFunction实现清洗逻辑

        // 3.2 DWD层-用户基础信息清洗（身高体重单位统一、异常值过滤）
        DataStream<DwdUserBasicInfoClean> dwdUserBasicInfoCleanStream = odsUserBasicInfoStream
                .map(new DwdUserBasicInfoCleanMap()); // 自定义MapFunction实现清洗逻辑

        // 4. DWS层：特征汇总计算（按用户聚合，遵循文档权重规则）
        // 4.1 DWS层-年龄特征汇总（1小时滚动窗口，按用户ID分组）
        KeyedStream<DwdUserBehaviorClean, String> ageKeyedStream = dwdUserBehaviorCleanStream
                .keyBy(new KeySelector<DwdUserBehaviorClean, String>() {
                    @Override
                    public String getKey(DwdUserBehaviorClean value) throws Exception {
                        return value.getUserId();
                    }
                });
        SingleOutputStreamOperator<DwdUserBehaviorClean> dwsUserAgeFeatureStream = ageKeyedStream
                .window(TumblingProcessingTimeWindows.of(Time.hours(1)))
                .reduce(new DwsUserAgeFeatureReduce()); // 自定义ReduceFunction实现得分汇总

        // 4.2 DWS层-性别特征汇总（逻辑类似年龄特征，省略重复代码）
        SingleOutputStreamOperator<DwdUserBehaviorClean> dwsUserGenderFeatureStream = ageKeyedStream
                .window(TumblingProcessingTimeWindows.of(Time.hours(1)))
                .reduce(new DwsUserGenderFeatureReduce());

        // 4.3 DWS层-身高体重星座汇总（逻辑类似，省略重复代码）
        SingleOutputStreamOperator<DwdUserBasicInfoClean> dwsUserBodyFeatureStream = dwdUserBasicInfoCleanStream
                .keyBy(DwdUserBasicInfoClean::getUserId)
                .window(TumblingProcessingTimeWindows.of(Time.hours(1)))
                .reduce(new DwsUserBodyFeatureReduce());

        // 5. ADS层：最终标签生成（遵循文档标签判定规则）
        ConnectedStreams<DwdUserBehaviorClean, DwdUserBehaviorClean> adsDmpBasicFeatureStream = dwsUserAgeFeatureStream
                .connect(dwsUserGenderFeatureStream); // 多流关联生成最终标签

        // 6. 结果输出（写入Hive表，遵循文档更新频率：每日更新+实时设备同步）
        tableEnv.createTemporaryView("ads_dmp_basic_feature", (Table) adsDmpBasicFeatureStream);
        tableEnv.executeSql("INSERT INTO hive_catalog.dmp_user_profile.ads_dmp_basic_feature " +
                "SELECT userId, ageSegment, genderTag, heightCm, weightKg, constellation, updateTime, dataDate " +
                "FROM ads_dmp_basic_feature");

        // 执行Flink Job
        env.execute("DmpBasicFeatureFlinkJob");
    }

    // ------------------------------ 自定义处理逻辑实现 ------------------------------
    /**
     * DWD层用户行为清洗MapFunction：实现类目匹配、搜索词分类、设备属性判定
     */
    static class DwdUserBehaviorCleanMap implements MapFunction<OdsUserBehavior, DwdUserBehaviorClean> {
        @Override
        public DwdUserBehaviorClean map(OdsUserBehavior value) throws Exception {
            DwdUserBehaviorClean clean = new DwdUserBehaviorClean();
            clean.setUserId(value.getUserId());
            clean.setBehaviorType(value.getBehaviorType());

            // 1. 匹配商品类目名称（模拟关联商品信息表，文档中商品类目属性数据）
            String categoryName = getCategoryName(value.getCategoryId());
            clean.setCategoryName(categoryName);

            // 2. 搜索词分类（遵循文档搜索词分析规则）
            String searchKeywordType = getSearchKeywordType(value.getSearchKeyword());
            clean.setSearchKeywordType(searchKeywordType);

            // 3. 设备属性判定（是否高端设备、是否最新系统版本，文档设备信息规则）
            clean.setHighEndDevice(isHighEndDevice(value.getDeviceType(), value.getDeviceVersion()));
            clean.setLatestVersion(isLatestVersion(value.getDeviceVersion()));

//            clean.setBehaviorTime(value.getBehaviorTime());
            clean.setDt(value.getDt());
            return clean;
        }

        // 模拟根据类目ID获取类目名称（文档中类目偏好系数表）
        private String getCategoryName(String categoryId) {
            Map<String, String> categoryMap = new HashMap<>();
            categoryMap.put("cat1001", "潮流服饰");
            categoryMap.put("cat1002", "家居用品");
            categoryMap.put("cat1003", "健康食品");
            return categoryMap.getOrDefault(categoryId, "其他");
        }

        // 搜索词分类（遵循文档搜索词分析规则）
        private String getSearchKeywordType(String keyword) {
            if (keyword == null) return "其他";
            if (keyword.matches(".*潮流|新款|时尚.*")) return "时尚与潮流";
            if (keyword.matches(".*优惠|折扣|便宜.*")) return "性价比";
            if (keyword.matches(".*健康食品|保健品|养生.*")) return "健康与养生";
            if (keyword.matches(".*婴儿用品|儿童玩具|亲子.*")) return "家庭与育儿";
            if (keyword.matches(".*智能手机|电脑配件|智能设备.*")) return "科技与数码";
            if (keyword.matches(".*在线课程|书籍|技能提升.*")) return "学习与发展";
            return "其他";
        }

        // 判定是否高端设备（文档中高端设备定义：价格高、配置好）
        private int isHighEndDevice(String deviceType, String version) {
            if ("ios".equals(deviceType) && version.matches("iOS 18.*")) return 1;
            if ("android".equals(deviceType) && version.matches("Android 15.*")) return 1;
            return 0;
        }

        // 判定是否最新系统版本（文档设备信息规则）
        private int isLatestVersion(String version) {
            if (version.matches("iOS 18.*|Android 15.*")) return 1;
            return 0;
        }
    }

    /**
     * DWD层用户基础信息清洗MapFunction：身高体重单位统一、异常值过滤（文档规则）
     */
    static class DwdUserBasicInfoCleanMap implements MapFunction<OdsUserBasicInfo, DwdUserBasicInfoClean> {
        @Override
        public DwdUserBasicInfoClean map(OdsUserBasicInfo value) throws Exception {
            DwdUserBasicInfoClean clean = new DwdUserBasicInfoClean();
            clean.setUserId(value.getUserId());

            // 1. 生日清洗（仅保留yyyy-MM-dd格式，文档星座计算数据源规则）
            clean.setBirthDate(validateBirthDate(value.getBirthDate()));

            // 2. 身高清洗（单位统一为cm，过滤100-250cm异常值，文档身高标签规则）
            clean.setHeightDevice(validateHeight(value.getDeviceHeight()));
            clean.setHeightAuth(validateHeight(parseHeight(value.getInputHeight())));
            clean.setHeightOrder(validateHeight(parseHeight(value.getOrderHeight())));
            clean.setHeightForm(validateHeight(parseHeight(value.getFormHeight())));

            // 3. 体重清洗（单位统一为kg，过滤30-200kg异常值，文档体重标签规则）
            clean.setWeightDevice(validateWeight(value.getDeviceWeight()));
            clean.setWeightAuth(validateWeight(parseWeight(value.getInputWeight())));
            clean.setWeightOrder(validateWeight(parseWeight(value.getOrderWeight())));
            clean.setWeightForm(validateWeight(parseWeight(value.getFormWeight())));

//            clean.setUpdateTime(value.getUpdateTime());
            clean.setDt(value.getDt());
            return clean;
        }

        // 生日格式验证
        private String validateBirthDate(String birthDate) {
            if (birthDate != null && birthDate.matches("^\\d{4}-\\d{2}-\\d{2}$")) {
                return birthDate;
            }
            return null;
        }

        // 身高解析（处理无单位、cm单位）
        private Double parseHeight(String heightStr) {
            if (heightStr == null) return null;
            String numStr = heightStr.replaceAll("[^0-9.]", "");
            if (numStr.isEmpty()) return null;
            return Double.parseDouble(numStr);
        }

        // 身高异常值过滤（100-250cm，文档身高标签规则）
        private Double validateHeight(Double height) {
            if (height != null && height >= 100 && height <= 250) {
                return height;
            }
            return null;
        }

        // 体重解析（处理斤、g、kg单位，文档体重标签规则）
        private Double parseWeight(String weightStr) {
            if (weightStr == null) return null;
            String numStr = weightStr.replaceAll("[^0-9.]", "");
            if (numStr.isEmpty()) return null;
            double num = Double.parseDouble(numStr);

            if (weightStr.contains("斤")) {
                return num / 2; // 斤转kg
            } else if (weightStr.contains("g")) {
                return num / 1000; // g转kg
            } else {
                return num; // 默认kg
            }
        }

        // 体重异常值过滤（30-200kg，文档体重标签规则）
        private Double validateWeight(Double weight) {
            if (weight != null && weight >= 30 && weight <= 200) {
                return weight;
            }
            return null;
        }
    }

    /**
     * DWS层年龄特征汇总ReduceFunction：按文档权重计算各维度得分
     */
    static class DwsUserAgeFeatureReduce implements ReduceFunction<DwdUserBehaviorClean> {
        @Override
        public DwdUserBehaviorClean reduce(DwdUserBehaviorClean value1, DwdUserBehaviorClean value2) throws Exception {
            // 1. 类目偏好得分计算（权重30%，文档类目偏好系数表）
            Map<String, Double> categoryScore1 = calculateCategoryScore(value1.getCategoryName());
            Map<String, Double> categoryScore2 = calculateCategoryScore(value2.getCategoryName());
            Map<String, Double> categoryScore = mergeScoreMap(categoryScore1, categoryScore2, 0.3);

            // 2. 品牌偏好得分计算（权重20%，文档品牌偏好系数表）
            Map<String, Double> brandScore1 = calculateBrandScore(value1.getBrandName());
            Map<String, Double> brandScore2 = calculateBrandScore(value2.getBrandName());
            Map<String, Double> brandScore = mergeScoreMap(brandScore1, brandScore2, 0.2);

            // 3. 其他维度得分计算（价格敏感度、时间行为等，逻辑类似，遵循文档权重）
            // ... 省略重复计算逻辑

            // 4. 构建DwsUserAgeFeature并返回（省略对象构建代码）
            return value1; // 实际应返回合并后的结果，此处为简化代码
        }

        // 计算类目偏好得分（文档类目偏好系数表）
        private Map<String, Double> calculateCategoryScore(String categoryName) {
            Map<String, Double> scoreMap = new HashMap<>();
            switch (categoryName) {
                case "潮流服饰":
                    scoreMap.put("18-24", 0.9);
                    scoreMap.put("25-29", 0.8);
                    scoreMap.put("30-34", 0.6);
                    scoreMap.put("35-39", 0.4);
                    scoreMap.put("40-49", 0.2);
                    scoreMap.put("50以上", 0.1);
                    break;
                case "家居用品":
                    scoreMap.put("18-24", 0.2);
                    scoreMap.put("25-29", 0.4);
                    scoreMap.put("30-34", 0.6);
                    scoreMap.put("35-39", 0.8);
                    scoreMap.put("40-49", 0.9);
                    scoreMap.put("50以上", 0.7);
                    break;
                case "健康食品":
                    scoreMap.put("18-24", 0.1);
                    scoreMap.put("25-29", 0.2);
                    scoreMap.put("30-34", 0.4);
                    scoreMap.put("35-39", 0.6);
                    scoreMap.put("40-49", 0.8);
                    scoreMap.put("50以上", 0.9);
                    break;
                default:
                    // 其他类目默认得分0
                    scoreMap.put("18-24", 0.0);
                    scoreMap.put("25-29", 0.0);
                    scoreMap.put("30-34", 0.0);
                    scoreMap.put("35-39", 0.0);
                    scoreMap.put("40-49", 0.0);
                    scoreMap.put("50以上", 0.0);
            }
            return scoreMap;
        }

        // 合并得分Map并乘以权重
        private Map<String, Double> mergeScoreMap(Map<String, Double> map1, Map<String, Double> map2, double weight) {
            Map<String, Double> mergedMap = new HashMap<>();
            for (String age : map1.keySet()) {
                double sum = map1.get(age) + map2.get(age);
                mergedMap.put(age, sum * weight);
            }
            return mergedMap;
        }

        // 品牌偏好得分计算（文档品牌偏好系数表，逻辑类似类目得分，省略代码）
        private Map<String, Double> calculateBrandScore(String brandName) {
            // ... 实现逻辑
            return new HashMap<>();
        }
    }

    // 其他自定义Function（DwsUserGenderFeatureReduce、DwsUserBodyFeatureReduce、AdsFeatureMap等）
    // 逻辑均遵循文档规则，因代码篇幅限制，省略重复实现，核心思路与年龄特征一致
    static class DwsUserGenderFeatureReduce implements ReduceFunction<DwdUserBehaviorClean> {
        @Override
        public DwdUserBehaviorClean reduce(DwdUserBehaviorClean value1, DwdUserBehaviorClean value2) throws Exception {
            // 按文档性别标签规则计算女性得分、男性得分、家庭品类占比
            // ... 实现逻辑
            return value1;
        }
    }

    static class DwsUserBodyFeatureReduce implements ReduceFunction<DwdUserBasicInfoClean> {
        @Override
        public DwdUserBasicInfoClean reduce(DwdUserBasicInfoClean value1, DwdUserBasicInfoClean value2) throws Exception {
            // 按文档身高体重权重规则计算加权值，解析星座
            // ... 实现逻辑
            return value1;
        }
    }



    // DWD层实体类定义（因篇幅限制，简化属性，实际需包含所有清洗后字段）
    @Data
    static class DwdUserBehaviorClean {
        private String userId;
        private String behaviorType;
        private String categoryName;
        private String brandName;
        private String priceRange;
        private String searchKeywordType;
        private int highEndDevice;
        private int latestVersion;
        private Timestamp behaviorTime;
        private String dt;
    }

    @Data
    static class DwdUserBasicInfoClean {
        private String userId;
        private String birthDate;
        private Double heightDevice;
        private Double heightAuth;
        private Double heightOrder;
        private Double heightForm;
        private Double weightDevice;
        private Double weightAuth;
        private Double weightOrder;
        private Double weightForm;
        private Timestamp updateTime;
        private String dt;
    }

    @Data
    static class DwsUserGenderFeature {
        private String userId;
        private Double femaleScore;
        private Double maleScore;
        private Double familyPurchaseRatio;
        private String statDate;
    }

    @Data
    static class DwsUserBodyFeature {
        private String userId;
        private Double weightedHeight;
        private Double weightedWeight;
        private String constellation;
        private String statDate;
    }
}